package com.bigdata.spark.streaming

import org.apache.spark.SparkConf
import org.apache.spark.streaming.dstream.ReceiverInputDStream
import org.apache.spark.streaming.{Seconds, StreamingContext}

/**
 * @author Gerry chan
 * @version 1.0
 * 2020/12/27
 * spark streaming https://www.bilibili.com/video/BV11A411L7CK?p=187
 * windows下执行：nc -lp 9999
 */
object SparkStreaming01_WordCount {
  def main(args: Array[String]): Unit = {
    //1.初始化spark配置信息
    val sparkConf = new SparkConf().setMaster("local[*]").setAppName("SparkStreaing")
    //2.初始化 SparkStreamingContext
    // 第二个参数表示批量数据处理的周期
    val ssc = new StreamingContext(sparkConf, Seconds(3))

    //3.通过监控端口 创建DStream,读进来的数据为一行行
    val lineStreams: ReceiverInputDStream[String] = ssc.socketTextStream("localhoust", 9999)

    val wordStreams = lineStreams.flatMap(_.split(" "))

    val wordAndOneStreams = wordStreams.map((_, 1))
    wordAndOneStreams.print()

    //3.启动SparkStreamingContext
    ssc.start()
    ssc.awaitTermination()
    //ssc.stop()
  }

}
